package com.abyss.sink.stream;


import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 将流数据写出到hdfs和本地文件和csv中
 */
public class FileSinkDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. Source
        DataStreamSource<Tuple3<Integer, String, Double>> source = env.fromElements(
                Tuple3.of(19, "潇潇", 170.50),
                Tuple3.of(11, "甜甜", 168.8),
                Tuple3.of(16, "刚刚", 178.8),
                Tuple3.of(19, "蛋蛋", 179.99)
        );

        // 3. sink
        // 3.1 Sink to local file
        source.writeAsText("data/output/xxx3", FileSystem.WriteMode.OVERWRITE).setParallelism(1);

        // 3.2 Sink to hdfs file
        source.writeAsText("hdfs://node1:8020/output/xxx3", FileSystem.WriteMode.OVERWRITE).setParallelism(1);

        // 3.3 Sink to local csv file
        source.writeAsCsv(
                "data/output/xxx3csv",
                FileSystem.WriteMode.OVERWRITE,
                "\n",
                ","
        );

        // 4. 执行
        env.execute();
    }
}